package com.datagateway.component;

import com.datagateway.model.ProcessedData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.*;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.Date;

/**
 * Hive数据源实现
 * 
 * @author Data Gateway Team
 * @version 1.0.0
 */
public class HiveDataSource implements DataSourceManager.DataSourceInstance {

    private static final Logger logger = LoggerFactory.getLogger(HiveDataSource.class);

    private final Map<String, Object> config;
    private final String type = "HIVE";
    private Connection connection;
    private PreparedStatement preparedStatement;
    private boolean connected = false;

    public HiveDataSource(Map<String, Object> config) {
        this.config = new ConcurrentHashMap<>(config);
        initialize();
    }

    /**
     * 初始化Hive连接
     */
    private void initialize() {
        try {
            String url = (String) config.get("url");
            String username = (String) config.get("username");
            String password = (String) config.get("password");
            String driver = (String) config.getOrDefault("driver", "org.apache.hive.jdbc.HiveDriver");

            // 加载驱动
            Class.forName(driver);

            // 设置连接属性
            Properties props = new Properties();
            props.setProperty("user", username);
            props.setProperty("password", password);
            props.setProperty("hive.server2.transport.mode", "binary");
            props.setProperty("hive.server2.thrift.sasl.qop", "auth");

            // 建立连接
            connection = DriverManager.getConnection(url, props);
            connected = true;

            logger.info("Hive数据源初始化成功: {}", url);

        } catch (Exception e) {
            logger.error("Hive数据源初始化失败: {}", e.getMessage(), e);
            connected = false;
        }
    }

    @Override
    public boolean testConnection() {
        try {
            if (connection == null || connection.isClosed()) {
                initialize();
            }

            if (connection != null && !connection.isClosed()) {
                try (Statement stmt = connection.createStatement()) {
                    stmt.executeQuery("SELECT 1");
                    connected = true;
                    return true;
                }
            }
            return false;

        } catch (SQLException e) {
            logger.error("Hive连接测试失败: {}", e.getMessage(), e);
            connected = false;
            return false;
        }
    }

    @Override
    public boolean writeData(List<ProcessedData> dataList) {
        if (!connected || connection == null) {
            logger.error("Hive连接未建立");
            return false;
        }

        if (dataList == null || dataList.isEmpty()) {
            return true;
        }

        String tableName = (String) config.getOrDefault("table", "processed_data");
        String insertSql = buildInsertSql(tableName);

        try {
            connection.setAutoCommit(false);
            preparedStatement = connection.prepareStatement(insertSql);

            int batchCount = 0;
            for (ProcessedData data : dataList) {
                setPreparedStatementParameters(data);
                preparedStatement.addBatch();
                batchCount++;

                // 批量提交
                if (batchCount % 1000 == 0) {
                    preparedStatement.executeBatch();
                    connection.commit();
                    batchCount = 0;
                }
            }

            // 提交剩余数据
            if (batchCount > 0) {
                preparedStatement.executeBatch();
                connection.commit();
            }

            logger.info("Hive数据写入成功: count={}", dataList.size());
            return true;

        } catch (SQLException e) {
            logger.error("Hive数据写入失败: {}", e.getMessage(), e);
            try {
                connection.rollback();
            } catch (SQLException rollbackEx) {
                logger.error("Hive回滚失败: {}", rollbackEx.getMessage(), rollbackEx);
            }
            return false;
        } finally {
            try {
                if (preparedStatement != null) {
                    preparedStatement.close();
                }
                connection.setAutoCommit(true);
            } catch (SQLException e) {
                logger.error("Hive清理资源失败: {}", e.getMessage(), e);
            }
        }
    }

    @Override
    public List<ProcessedData> readData(String query, Map<String, Object> parameters) {
        if (!connected || connection == null) {
            logger.error("Hive连接未建立");
            return Collections.emptyList();
        }

        List<ProcessedData> resultList = new ArrayList<>();

        try {
            preparedStatement = connection.prepareStatement(query);

            // 设置参数
            if (parameters != null) {
                int paramIndex = 1;
                for (Map.Entry<String, Object> entry : parameters.entrySet()) {
                    preparedStatement.setObject(paramIndex++, entry.getValue());
                }
            }

            try (ResultSet resultSet = preparedStatement.executeQuery()) {
                while (resultSet.next()) {
                    ProcessedData data = mapResultSetToProcessedData(resultSet);
                    resultList.add(data);
                }
            }

            logger.info("Hive数据读取成功: count={}", resultList.size());
            return resultList;

        } catch (SQLException e) {
            logger.error("Hive数据读取失败: {}", e.getMessage(), e);
            return Collections.emptyList();
        } finally {
            try {
                if (preparedStatement != null) {
                    preparedStatement.close();
                }
            } catch (SQLException e) {
                logger.error("Hive清理资源失败: {}", e.getMessage(), e);
            }
        }
    }

    @Override
    public boolean executeSql(String sql, Map<String, Object> parameters) {
        if (!connected || connection == null) {
            logger.error("Hive连接未建立");
            return false;
        }

        try {
            preparedStatement = connection.prepareStatement(sql);

            // 设置参数
            if (parameters != null) {
                int paramIndex = 1;
                for (Map.Entry<String, Object> entry : parameters.entrySet()) {
                    preparedStatement.setObject(paramIndex++, entry.getValue());
                }
            }

            boolean result = preparedStatement.execute();
            logger.info("Hive SQL执行成功: {}", sql);
            return result;

        } catch (SQLException e) {
            logger.error("Hive SQL执行失败: sql={}, error={}", sql, e.getMessage(), e);
            return false;
        } finally {
            try {
                if (preparedStatement != null) {
                    preparedStatement.close();
                }
            } catch (SQLException e) {
                logger.error("Hive清理资源失败: {}", e.getMessage(), e);
            }
        }
    }

    @Override
    public boolean isConnected() {
        try {
            return connected && connection != null && !connection.isClosed();
        } catch (SQLException e) {
            return false;
        }
    }

    @Override
    public void close() {
        try {
            if (preparedStatement != null) {
                preparedStatement.close();
            }
            if (connection != null && !connection.isClosed()) {
                connection.close();
            }
            connected = false;
            logger.info("Hive数据源连接已关闭");
        } catch (SQLException e) {
            logger.error("关闭Hive连接失败: {}", e.getMessage(), e);
        }
    }

    @Override
    public String getType() {
        return type;
    }

    @Override
    public Map<String, Object> getConfig() {
        return new HashMap<>(config);
    }

    /**
     * 构建插入SQL语句
     */
    private String buildInsertSql(String tableName) {
        return String.format(
            "INSERT INTO TABLE %s PARTITION (dt='%s') VALUES (?, ?, ?, ?)",
            tableName, getCurrentDate()
        );
    }

    /**
     * 获取当前日期字符串
     */
    private String getCurrentDate() {
        return new java.text.SimpleDateFormat("yyyyMMdd").format(new Date());
    }

    /**
     * 设置PreparedStatement参数
     */
    private void setPreparedStatementParameters(ProcessedData data) throws SQLException {
        preparedStatement.setString(1, data.getId());
        preparedStatement.setString(2, data.getContent());
        preparedStatement.setLong(3, data.getTimestamp());
        
        // 将metadata转换为JSON字符串
        String metadataJson = convertMetadataToJson(data.getMetadata());
        preparedStatement.setString(4, metadataJson);
    }

    /**
     * 将ResultSet映射为ProcessedData
     */
    private ProcessedData mapResultSetToProcessedData(ResultSet resultSet) throws SQLException {
        ProcessedData data = new ProcessedData();
        data.setId(resultSet.getString("id"));
        data.setContent(resultSet.getString("content"));
        data.setTimestamp(resultSet.getLong("timestamp"));
        
        // 解析metadata JSON
        String metadataJson = resultSet.getString("metadata");
        Map<String, Object> metadata = parseMetadataFromJson(metadataJson);
        data.setMetadata(metadata);
        
        return data;
    }

    /**
     * 将metadata转换为JSON字符串
     */
    private String convertMetadataToJson(Map<String, Object> metadata) {
        if (metadata == null || metadata.isEmpty()) {
            return "{}";
        }
        
        try {
            // 简单的JSON转换，实际项目中可以使用Jackson等库
            StringBuilder json = new StringBuilder("{");
            boolean first = true;
            for (Map.Entry<String, Object> entry : metadata.entrySet()) {
                if (!first) {
                    json.append(",");
                }
                json.append("\"").append(entry.getKey()).append("\":");
                if (entry.getValue() instanceof String) {
                    json.append("\"").append(entry.getValue()).append("\"");
                } else {
                    json.append(entry.getValue());
                }
                first = false;
            }
            json.append("}");
            return json.toString();
        } catch (Exception e) {
            logger.error("转换metadata为JSON失败: {}", e.getMessage(), e);
            return "{}";
        }
    }

    /**
     * 从JSON字符串解析metadata
     */
    private Map<String, Object> parseMetadataFromJson(String metadataJson) {
        Map<String, Object> metadata = new HashMap<>();
        if (metadataJson == null || metadataJson.trim().isEmpty() || "{}".equals(metadataJson)) {
            return metadata;
        }
        
        try {
            // 简单的JSON解析，实际项目中可以使用Jackson等库
            metadataJson = metadataJson.trim();
            if (metadataJson.startsWith("{") && metadataJson.endsWith("}")) {
                metadataJson = metadataJson.substring(1, metadataJson.length() - 1);
                String[] pairs = metadataJson.split(",");
                for (String pair : pairs) {
                    String[] keyValue = pair.split(":");
                    if (keyValue.length == 2) {
                        String key = keyValue[0].trim().replaceAll("\"", "");
                        String value = keyValue[1].trim().replaceAll("\"", "");
                        metadata.put(key, value);
                    }
                }
            }
        } catch (Exception e) {
            logger.error("解析metadata JSON失败: {}", e.getMessage(), e);
        }
        
        return metadata;
    }
}
